package com.atguigu.wc;/**
 * Copyright (c) 2018-2028 尚硅谷 All Rights Reserved
 * <p>
 * Project: FlinkTutorial
 * Package: com.atguigu.wc
 * Version: 1.0
 * <p>
 * Created by wushengran on 2020/11/6 11:48
 */

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.net.URL;

/**
 * @ClassName: StreamWordCount
 * @Description:
 * @Author: wushengran on 2020/11/6 11:48
 * @Version: 1.0
 */
public class StreamWordCount {


    /**
     * setParallelism优先级顺序（高->低）：1.api代码中配置-> 2.env.setParallelism(1) -> 3.admin仪表盘中配置 -> 4.yml配置文件中配置
     *
     *
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception{
        // 创建流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 会自动判断是否为本地还是远程
//        env.setParallelism(1);
//        env.disableOperatorChaining();// 关闭任务链，每个任务独立执行

//        // 从文件中读取数据
//        String inputPath = "D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\hello.txt";
//        String inputPath = WordCount.class.getClassLoader().getResource("hello.txt").getPath();
//        DataStream<String> inputDataStream = env.readTextFile(inputPath);

        // 用parameter tool工具从程序启动参数中提取配置项

        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String host = parameterTool.get("host");
        int port = parameterTool.getInt("port");

        // 从socket文本流读取数据
        DataStream<String> inputDataStream = env.socketTextStream(host, port);


        // 基于数据流进行转换计算
        DataStream<Tuple2<String, Integer>> resultStream = inputDataStream.flatMap(new WordCount.MyFlatMapper()).slotSharingGroup("green")
                .keyBy(0)// 0- 第一个参数位置
                .sum(1).setParallelism(2).slotSharingGroup("red").shuffle().iterate().startNewChain();// 指定不同的sharingGroup，这样不在同一个slot中执行

        // startNewChain(); 不指定任务的合并，分开执行
        resultStream.print().setParallelism(1);// 很灵活每一步都可以设置Parallelism

        // 执行任务，因为是流失任务，要execute执行
        env.execute();
    }
}
